package com.codeyang.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.codeyang.constant.QueueConstant;
import com.codeyang.dao.ProdEsDao;
import com.codeyang.domain.Prod;
import com.codeyang.es.ProdEs;
import com.codeyang.service.ImportService;
import com.codeyang.service.ProdService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * 描述: TO-删除标志-DO
 *
 * @author CodeYang_Site
 * @version 2021/5/25 23:04
 */
@Slf4j
@Service
public class ImportServiceImpl implements ImportService, CommandLineRunner {


    /**
     * 页大小
     */
    @Value("${esimport.size}")
    private Integer size;

    @Autowired
    private ProdService prodService;

    @Autowired
    private ProdEsDao prodEsDao;

    private Date start;

    /**
     * 全量导入
     * mysql -> 同步到es 大数据量如 500w
     * 项目一启动执行的
     * step: 分批次 + 多线程 分页  控制页码 页大小
     */
    @Override
    public void importAll() {
        log.info("全量导入开始");
        //1.查询商品总数量 用于计算页码数量
        Integer total = prodService.getTotal(null, null);

        if (total == null || total.equals(0)) {
            log.info("total 等于 {} 没有商品", total);
            start = new Date();
            return;
        }
        //2.有就计算页码数 固定公司 总条数%页大小 ==0 ? 总条数/页带下 : 总条数/页大小 +1
        int totalPage = total % size == 0 ? total / size : total / size + 1;
        //3.循环 totalPage 来 分页查询一批数据导入es
        for (int i = 1; i <= totalPage; i++) {
            //导入es的方法
            importEs(i, size, null, null);
        }
        //记录最开始的时间
        start = new Date();
        log.info("全量导入结束");
    }

    /**
     * 导入es的方法
     *
     * @param current
     * @param size
     * @param start
     * @param end
     */
    private void importEs(int current, Integer size, Date start, Date end) {
        //1.优化查询步骤发一个sql搞定
        Page<Prod> prodPage = new Page<>(current, size, false);
        //查询分页 得到MSYQL数据
        List<Prod> prodList = prodService.findEsProdPage(prodPage, start, end);
        //转换 prodEs集合
        List<ProdEs> prodEs = new ArrayList<>(prodList.size());
        prodList.forEach(prod -> {
            ProdEs prodEsnow = new ProdEs();
            //拷贝属性 忽略大小写
            BeanUtil.copyProperties(prod, prodEsnow, true);

            prodEs.add(prodEsnow);
        });
        //导|写入Es
        prodEsDao.saveAll(prodEs);

    }

    /**
     * 增量导入
     *
     * @Scheduled 定时任务支持
     * cron 表达式 如每天上午十点 开始执行
     * # 每天 5 点半
     * 0 30 5 * * ?
     * initialDelay 首次执行的时间
     * fixedRate  首次执行以后 每隔多少毫秒执行一次
     * 2分钟后执行这个定时任务,而且每过两分钟执行一次
     * 测试OK 改为每半个小时干一次
     */
    @Override
    @Scheduled(initialDelay = 180 * 1000, fixedDelay = 180 * 1000)
    public void importUpdate() {
        log.info("增量导入开始");

        Date end = new Date();

        //1.查询商品总数量 用于计算页码数量
        Integer total = prodService.getTotal(start, end);
        if (total == null || total.equals(0)) {
            log.info("total 等于 {} 没有商品", total);
            log.info("增量导入结束");
            start = new Date();
            return;
        }
        //2.有就计算页码数 固定公司 总条数%页大小 ==0 ? 总条数/页带下 : 总条数/页大小 +1
        int totalPage = total % size == 0 ? total / size : total / size + 1;
        //3.循环 totalPage 来 分页查询一批数据导入es
        for (int i = 1; i <= totalPage; i++) {
            //导入es的方法
            importEs(i, size, start, end);
        }
        // 这样一致把 当前任务的结束时间交给 start end每次是定时任务开始的最新的,形成一个时间窗口滑动,且固定了时间范围
        start = end;
        log.info("增量导入结束");
    }

    /**
     * 快速导入 -- 实时导入
     * 有些业务数据,需要立马将MYSQL的数据导入到ES中 vip数据等
     * step:
     * 1.借助MQ 帮我们完成
     * 2.如 秒杀,减少库存
     * 3.下单完成之后 往MQ里放一个消息  监听干这个 (减MYSQL 库存,es)
     * 如: 一个订单  包含多个商品 2      Map<Long,Integer > prodId   count
     * 超过30分钟不支付  订单要回滚
     */
    @Override
    public void importQuick() {

    }


    /**
     * 实时队列MQ处理
     * 如: 监听商品的修改|实时提交队列任务
     *  监听 Es修改队列
     * @param message
     * @param channel
     */
    //concurrency 开启几个线程去处理数据
    @RabbitListener(queues = QueueConstant.PROD_CHANGE_QUEUE,concurrency ="3-5" )
    public void handlerProdEsChange(Message message, Channel channel) {
        //1.拿到投递id
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        log.info("有修改商品消息进入");
        //消息的处理 与 放入消息有关 要设计好消息的结构 一般直接使用 id等标识属性
        //如 {1:2} {1:-2} id为1的商品新增两个 {prodID,count}
        String msg = new String(message.getBody());

        JSONObject jsonObject = JSON.parseObject(msg);
        // 2拿到所有的prodId
        List<Long> prodIds = new ArrayList<>();

        // 循环这个集合 这个jsonObject ----这里就要请我们 存的那个map必须是 Map<proId,count> msg
        jsonObject.forEach((prodId, count) -> {
            prodIds.add(Long.valueOf(prodId));
        });

        // 查询es  因为要实时修改es
        Iterable<ProdEs> prodEsList = prodEsDao.findAllById(prodIds);

        prodEsList.forEach(prodEs -> {
            Integer count = jsonObject.getInteger(prodEs.getProdId().toString());
            // 设置  这里通过 +- 号判断是否是加还是减库存
            long finalCount = prodEs.getTotalStocks() + count;
            if (finalCount < 0) {
                log.error("id为{}的商品库存不足" + prodEs.getProdId());
                // 记录数据库之类的操作
                try {
                    channel.basicAck(deliveryTag, false);
                } catch (IOException e) {
                    e.printStackTrace();
                }
                return;
            }
            prodEs.setTotalStocks(finalCount);
            prodEs.setSoldNum(prodEs.getSoldNum() - count);
            //这里本质上还要修改MYSQL 操作库存和销量
        });
        try {
            // 插入
            prodEsDao.saveAll(prodEsList);
            // 签收
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("有处理修改商品消息完成");
    }

    /**
     * Callback used to run the bean.
     * 要想在spring 导入bean之后就调用方法两种操作
     * -- 这两种是在Ioc容器加载完毕的,可以直接使用ioc里的bean对象
     * 1. 在类里使用 @postConstructor 本质 --> 是一个代理的后置的 构造方法
     * 2.实习 CommandLineRunner接口 在这个方法里去调用
     *
     * @param args incoming main method arguments
     * @throws Exception on error
     */
    @Override
    public void run(String... args) throws Exception {
        //在spring初始化后调用 全量导入
        importAll();
    }
}
